package com.apps.sdses.flink141.demo;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author jiwei
 * @description
 * @date 2023/12/29 10:15
 */
public class TestCheck {
    public static void main(String[] args) throws Exception {
        //StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(1000);
        env.setParallelism(1);

        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        tEnv.executeSql(
                "CREATE TABLE kafka_ogg_source (\n" +
                        "  xhb_id string,\n" +
                        "  xhb_name string,\n" +
                        "  xhb_index int,\n" +
                        "  hs_id string,\n" +
                        "  dt TIMESTAMP(3),\n" +
                        "  xhbpl decimal(10,2),\n" +
                        "  xhbdl decimal(10,2)\n" +
                        ") WITH (\n" +
                        "  'connector' = 'kafka',\n" +
                        "  'topic' = 'SHRL_XHB',\n" +
                        "  'scan.startup.mode' = 'earliest-offset',\n" +
                        "  'properties.bootstrap.servers' = '173.18.162.33:29092',\n" +
                        "  'properties.group.id' = 'test_xhb_20231229',\n" +
                        "  'debezium-json.ignore-parse-errors' = 'true',\n" +
                        "  'format' = 'debezium-json'\n" +
                        ")");

        //doris table
        tEnv.executeSql(
                "CREATE TABLE doris_test_sink (" +
                        "  xhb_id string,\n" +
                        "  xhb_name string,\n" +
                        "  xhb_index int,\n" +
                        "  hs_id string,\n" +
                        "  dt TIMESTAMP(3),\n" +
                        "  xhbpl decimal(10,2),\n" +
                        "  xhbdl decimal(10,2)\n" +
                        ") " +
                        "WITH (\n" +
                        "  'connector' = 'doris',\n" +
                        "  'fenodes' = '173.18.162.50:38030',\n" +
                        "  'jdbc-url' = 'jdbc:mysql://173.18.162.50:16030',\n" +
                        "  'table.identifier' = 'db_datacube_doris_test.SHRL_XHB_TEMP',\n" +
                        "  'username' = 'datacube_doris_test',\n" +
                        "  'password' = 'Nx8Tz_Y8v3x',\n" +
                        "  'sink.label-prefix' = 'SHRL_XHB_TEMP_20231229_09',\n" +
                        "  'sink.properties.format' = 'json',\n" +       //json data format
                        "  'sink.enable-delete' = 'true',\n" +
                        "  'sink.batch.size' = '50000',\n" +
                        "  'sink.batch.interval' = '60000',\n" +       //json data format
                        "  'sink.enable-2pc' = 'false',\n" +
                        "  'sink.properties.read_json_by_line' = 'true'\n" +
                        ")");

        //insert into kafka ogg data to doris table
        tEnv.executeSql("INSERT INTO doris_test_sink select * from kafka_ogg_source");
    }
}
